import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.io.Serializable;

public class TopicConsumer {
	public static MessageConsumer createConsumer(ActiveMQConnectionFactory factory, String topic) throws JMSException {
		factory.setTrustAllPackages(true); // 设置信任所有包
		// 创建连接
		Connection connection = factory.createConnection();
		// 启动连接
		connection.start();
		// 创建会话
		Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		// 创建主题
		Topic sessionTopic = session.createTopic(topic);
		// 创建消息消费者
		return session.createConsumer(sessionTopic);
	}

	public static void main(String[] args) throws JMSException {
		// 创建连接工厂
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		createConsumer(factory, "asm").setMessageListener((message_asm) -> setMessageListener(message_asm, "asm"));
		createConsumer(factory, "se").setMessageListener((message_se) -> setMessageListener(message_se, "se"));
	}

	static String setMessageListener(Message message, String topic) {
		try {
			if (message instanceof ObjectMessage) {
				ObjectMessage objectMessage = (ObjectMessage) message;
				Serializable obj = objectMessage.getObject();
				if (obj instanceof MyMessage) {
					MyMessage myMessage = (MyMessage) obj;
					String content = "Topic " + topic + " Received message: " + myMessage.getContent();
					System.out.println(content);
					return content;
				}
			}
			message.acknowledge(); // 发送消息确认
		} catch (JMSException e) {
			e.printStackTrace();
		}
		return null;
	}
}